package cn.ymatrix.httpTOYMatrix.energyCar;

import cn.ymatrix.utils.Config;
import com.alibaba.fastjson.JSON;
import org.apache.commons.io.FileUtils;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.HttpClientBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class DataHandler {
  private static final Logger LOGGER = LoggerFactory.getLogger(DataHandler.class);

  private static List<MeasurementJSON> measurementSchemas;

  private static final List<String> devicePaths = new ArrayList<>();

  private static final int CORE_POOL_SIZE = 5;
  private static final long KEEP_ALIVE_TIME = 200L;

  public static void main(String[] args) {
    initDeviceAndMeasurement();
  }

  public static void initDeviceAndMeasurement() {
    File file = new File(Config.getEneryCarFile());

    try {
      measurementSchemas =
          JSON.parseArray(
              FileUtils.readFileToString(file, StandardCharsets.UTF_8), MeasurementJSON.class);

      String devicePathPrefix = Config.getDevicePathPrefix();
      int deviceStartNum = Config.getDeviceStartNum();

      for (int i = 0; i < Config.getDeviceNum(); i++) {
        String tmpPath = devicePathPrefix + (deviceStartNum + i);
        System.out.println("tmpPath:" + tmpPath);
        devicePaths.add(tmpPath);
      }

    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }

  public void insertDataByBatch(CountDownLatch countDownLatch) {
    long startTime = Config.getUnixMsStartTime();
    long endTime = Config.getUnixMsEndTime();
    int batchSize = Config.getBatchSize();
    int pointStep = Config.getPointStep();

    ThreadPoolExecutor threadPoolExecutor =
        new ThreadPoolExecutor(
            CORE_POOL_SIZE,
            Config.getMaximumPoolSize(),
            KEEP_ALIVE_TIME,
            TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>());

    HttpClient httpClient =
        HttpClientBuilder.create().setMaxConnPerRoute(Config.getMaxConnPerRoute()).build();

    while (startTime < endTime) {
      CountDownLatch deviceLatch = new CountDownLatch(devicePaths.size());
      long tmpStartTime = startTime;

      for (String devicePath : devicePaths) {
        threadPoolExecutor.submit(
            () -> sendHttpPostRequest(httpClient, devicePath, tmpStartTime, deviceLatch));
      }

      try {
        deviceLatch.await();
      } catch (InterruptedException e) {
        LOGGER.error("InterruptedException occurred while waiting for device latch.", e);
      }
      startTime += (long) batchSize * pointStep;
    }
    countDownLatch.countDown();
    LOGGER.info(
        "insertTablet finish, startTime: {}, endTime: {}, pointStep: {}",
        Config.getUnixMsEndTime(),
        Config.getUnixMsEndTime(),
        Config.getPointStep());
  }

  private void sendHttpPostRequest(
      HttpClient httpClient, String devicePath, long startTime, CountDownLatch deviceLatch) {
    HttpPost httpPost = new HttpPost(Config.getHttpUrl());
    try {
      httpPost.setEntity(
          new StringEntity(getPostJsonData(devicePath, startTime), StandardCharsets.UTF_8));
      httpPost.setHeader("Content-Type", "text/plain");

      httpClient.execute(httpPost);

      System.out.println("----------------");

    } catch (IOException e) {
      e.printStackTrace();
    } finally {
      httpPost.releaseConnection();
      deviceLatch.countDown();
    }
  }

  private String getPostJsonData(String devicePath, long startTime) {
    StringBuilder dataStringBuilder = new StringBuilder(Config.getTableName() + "\n");

    for (int rowIndex = 0; rowIndex < Config.getBatchSize(); rowIndex++) {
      long time = startTime + (long) rowIndex * Config.getPointStep();
      if (time > Config.getUnixMsEndTime()) {
        break;
      }

      dataStringBuilder.append(time / 1000).append("|");
      dataStringBuilder.append(devicePath).append("|");

      for (MeasurementJSON measurementSchema : measurementSchemas) {
        switch (measurementSchema.getDataType()) {
          case DataConstant.INT32:
            dataStringBuilder.append(DataUtil.getIntValue()).append("|");
            break;
          case DataConstant.INT64:
            dataStringBuilder.append(DataUtil.getLongValue()).append("|");
            break;
          case DataConstant.FLOAT:
            dataStringBuilder.append(DataUtil.getFloatValue()).append("|");
            break;
          case DataConstant.DOUBLE:
            dataStringBuilder.append(DataUtil.getDoubleValue()).append("|");
            break;
          default:
            break;
        }
      }
      dataStringBuilder.deleteCharAt(dataStringBuilder.length() - 1);
      dataStringBuilder.append("\n");
    }
    // System.out.println("---"+dataStringBuilder.toString());
    return dataStringBuilder.toString();
  }
}
